/**
@file
	PollMessageDevice.cpp
@brief
	Implentation of PollMessageDevice.
@author
	GengYong
@revision
	2005/11/27 * Created by GengYong.
*/

#include "PollMessageDevice.h"

IMPLEMENT_MESSAGE_ROUTER(EPollMessageRouter, "Free", "TCP message router base on epoll.");

#include <cassert>
#include <unistd.h>
#include <errno.h>
#include <sys/socket.h>
#include <arpa/inet.h>
//----------------------------------------------------------------------------
//	Debugger
//----------------------------------------------------------------------------
#ifdef _DEBUG
#define LOG(...) //this->m_hostKernel->OutputConsole
#else
#define LOG(...)
#endif

//----------------------------------------------------------------------------
//	Internal Header
//----------------------------------------------------------------------------
namespace Internal
{			
	typedef struct PackageHeader
	{
		unsigned short length;
		unsigned short routine;	
		unsigned short catalog;
		unsigned short command;
	} PackageHeader;
}

//----------------------------------------------------------------------------
//	PollMessageDevice class
//----------------------------------------------------------------------------
bool EPollMessageRouter::Init(SAF::IKernel * kernel, SAF::IConfig * config)
{
	PERFCOUNTER("PollMessageDevice::Init");

	try
	{
		unsigned short port = (unsigned short)(atoi(config->GetValue("PORT")));
		if (0 == port) 
		{
			kernel->OutputConsole("[POLL] Error: Port not defined.");
			return false;
		}

		this->m_recvBufLength	= atoi(config->GetValue("RECVBUFLENGTH"));
		this->m_sendBufLength	= atoi(config->GetValue("SENDBUFLENGTH"));
		this->m_maxConnCount	= atoi(config->GetValue("MAXCONNECTION"));	

		if (this->m_recvBufLength == 0)
		{
			this->m_recvBufLength = 0x10000;
			kernel->OutputConsole("[POLL] Warnning: RECVBUFLENGTH not defined, set to default value %d.", this->m_recvBufLength);
		}

		if (this->m_sendBufLength == 0)
		{
			this->m_sendBufLength = 0x10000;
			kernel->OutputConsole("[POLL] Warnning: SENDBUFLENGTH not defined, set to default value %d.", this->m_sendBufLength);
		}

		if (this->m_maxConnCount == 0)
		{
			this->m_maxConnCount = 1024;
			kernel->OutputConsole("[POLL] Warnning: MAXCONNECTION not defined, set to default value %d.", this->m_maxConnCount);
		}

		if (strlen(config->GetValue("WAITTIMEOUT")) == 0)
		{
			this->m_waitTimeOut = 0;
			kernel->OutputConsole("[POLL] Warnning: WAITTIMEOUT not defined, set to default value %d.", this->m_waitTimeOut);
		}
		else
		{
			this->m_waitTimeOut = atoi(config->GetValue("WAITTIMEOUT"));
		}

		try
		{
			this->m_eventBuffer = new epoll_event[this->m_maxConnCount];
			this->m_msgDescStack = new MessageDescriptor[this->m_maxConnCount];
			this->m_profileBase = new ProfileList[this->m_maxConnCount];

			// init profile buffer and init table.
			memset(this->m_profileBase, 0, sizeof(ProfileList) * this->m_maxConnCount);
			for (int i = 0; i < this->m_maxConnCount; i++)
			{
				this->m_profileBase[i].recvbuf = new char[this->m_recvBufLength + sizeof(Internal::PackageHeader)];
				this->m_profileBase[i].sendbuf = new char[this->m_sendBufLength + sizeof(Internal::PackageHeader)];	
				this->m_profileBase[i].recv_ptr = this->m_profileBase[i].recvbuf;
				this->m_profileBase[i].send_ptr = this->m_profileBase[i].sendbuf;
				this->m_profileBase[i].proc_ptr = this->m_profileBase[i].recvbuf;
				this->m_profileBase[i].recv_left = this->m_recvBufLength;
				this->m_profileBase[i].send_left = this->m_sendBufLength;
			}

			for (int i = 0; i < this->m_maxConnCount - 1; i++)
			{
				this->m_profileBase[i].next = &(this->m_profileBase[i + 1]);
			}

			this->m_rootProfile = &(this->m_profileBase[0]);
			this->m_profileHead = &(this->m_profileBase[1]);
			this->m_profileTail = &(this->m_profileBase[this->m_maxConnCount - 1]);			
		}
		catch(const std::bad_alloc&)
		{
			kernel->OutputConsole("[POLL] Error: No enough memory.");
			throw std::bad_alloc();
		}

		
		this->m_rootProfile->sock = socket(AF_INET, SOCK_STREAM, 0);
		if (this->m_rootProfile->sock == -1)
		{		
			kernel->OutputConsole("[POLL] Error: Create listen socket failed.");
			throw std::bad_alloc();
		}

		sockaddr_in addr;
		memset(&addr, 0, sizeof(addr));
		addr.sin_family = PF_INET;
		addr.sin_addr.s_addr = inet_addr(config->GetValue("IP"));
		addr.sin_port = htons(port);
		if (bind(this->m_rootProfile->sock, (const sockaddr *)&addr, sizeof(addr)) == -1)
		{
			kernel->OutputConsole("[POLL] Error: Bind listen socket failed. (errcode: %d)", errno);
			throw std::bad_alloc();
		}

		if (listen(this->m_rootProfile->sock, 5) == -1)
		{
			kernel->OutputConsole("[POLL] Error: Listen failed. (errcode: %d)", errno);
			throw std::bad_alloc();
		}

		this->m_pollHandle = epoll_create(this->m_maxConnCount);
		if (m_pollHandle < 0)
		{
			kernel->OutputConsole("[POLL] Error: Create epoll failed.");
			throw std::bad_alloc();
		}

		epoll_event event;	
		this->m_rootProfile->conn = 0;
		this->m_rootProfile->key = NULL;
		event.data.ptr	= this->m_rootProfile;
		event.events	= EPOLLIN | EPOLLERR;		
		if (epoll_ctl(this->m_pollHandle, EPOLL_CTL_ADD, this->m_rootProfile->sock, &event) < 0)
		{
			kernel->OutputConsole("[POLL] Error: Add listen socket to epoll failed.");
			throw std::bad_alloc();
		}
		
		kernel->OutputConsole("[POLL] POLL message device listen at addr:%s:%d", 
			config->GetValue("IP"), port);

		this->m_msgDescCount	= 0;
		this->m_msgDescCursor	= 0;
		this->m_connections		= 1;
		this->m_hostKernel		= kernel;
		
		return true;
	}
	catch(const std::bad_alloc&)
	{		
		this->Release();		
	}
	return false;
}

void EPollMessageRouter::Release()
{
	if (this->m_pollHandle != -1)
	{
		close(this->m_pollHandle);
		this->m_pollHandle = -1;
	}

	if (this->m_profileBase != NULL)
	{
		for (int i = 0; i < this->m_maxConnCount; i++)
		{
			if (this->m_profileBase[i].sock != -1)
			{
				close(this->m_profileBase[i].sock);
			}
			if (this->m_profileBase[i].recvbuf != NULL)
			{
				delete[] this->m_profileBase[i].recvbuf;
			}
			if (this->m_profileBase[i].sendbuf != NULL)
			{
				delete[] this->m_profileBase[i].sendbuf;
			}
		}
		delete[] this->m_profileBase;
		this->m_profileHead = NULL;
		this->m_profileTail = NULL;
	}

	if (this->m_msgDescStack != 0)
	{
		delete[] this->m_msgDescStack;
		this->m_msgDescStack = NULL;
	}

	if (this->m_eventBuffer != 0)
	{
		delete[] this->m_eventBuffer;
		this->m_eventBuffer = NULL;
	}
	
	this->m_hostKernel		= NULL;
	this->m_connections		= 0;
	this->m_msgDescCount	= 0;
	this->m_msgDescCursor	= 0;
	this->m_recvBufLength	= 0;
	this->m_sendBufLength	= 0;
	this->m_maxConnCount	= 0;	

	return;
}


//bool PollMessageRouter::SendNetMessage(unsigned long nodeid, const SAF::IMessageRouter::NetMessage& message, const SAF::IMessageRouter::NetBuffer * buffers, unsigned long bufcount)
bool EPollMessageRouter::SendNetMessage(const SAF::IConnection * conn, const SAF::NetMessage& message, const SAF::NetBuffer * buffers, unsigned int bufcount)
{
	assert(!((bufcount != 0) && (buffers == NULL)));

	PERFCOUNTER("PollMessageDevice::SendNetMessage");

	int append_bytes = 0;
	for (unsigned long i = 0; i < bufcount; i++)
	{
		assert((buffers[i].data != NULL && buffers[i].size > 0) || (buffers[i].data == NULL && buffers[i].size == 0));		
		append_bytes += buffers[i].size;
	}

	assert(append_bytes < 65536);

	ProfileList * profile = &(this->m_profileBase[conn->Id()]);

	//assert(profile->nodeid == nodeid);

	//@ !!
	//=================================================================================
	if (profile->send_ptr == profile->sendbuf)
	{
		epoll_event event;
		event.data.ptr = profile;
		event.events = EPOLLIN | EPOLLOUT | EPOLLERR;
		epoll_ctl(this->m_pollHandle, EPOLL_CTL_MOD, profile->sock, &event);
	}
	//=================================================================================

	if (profile->send_left >= int(sizeof(Internal::PackageHeader)) + append_bytes)
	{	
		Internal::PackageHeader * header = (Internal::PackageHeader *)profile->send_ptr;
		header->length = (unsigned short)(append_bytes);
		header->command = message.command;
		header->catalog = message.catalog;
		header->routine = 0;		
		
		profile->send_ptr	+= sizeof(Internal::PackageHeader);

		for (unsigned long i = 0; i < bufcount; i++)
		{
			memcpy(profile->send_ptr, buffers[i].data, buffers[i].size);
			profile->send_ptr += buffers[i].size;
		}

		profile->send_left -= sizeof(Internal::PackageHeader) + append_bytes;
		
		//------------------------------
		//save the flux_in information.
		//profile->flux_out_packets++;
		//this->m_FluxOutPackets++;
		//------------------------------		
		return true;
	}

	assert(!"buffer overflow");

	return false;
}


//bool PollMessageRouter::RecvNetMessage(unsigned long& nodeid, SAF::IMessageRouter::NetPackage& package, void ** lpkey, unsigned long time)
bool EPollMessageRouter::RecvNetMessage(SAF::IConnection *& conn, SAF::NetPackage& package, unsigned int time)
{
	PERFCOUNTER("PollMessageDevice::RecvNetMessage");

	if (this->m_msgDescCursor < this->m_msgDescCount)
	{
		conn = this->m_msgDescStack[this->m_msgDescCursor].profile->conn;
		package = this->m_msgDescStack[this->m_msgDescCursor].package;
		//if (lpkey != NULL)
		//{
		//	*lpkey = this->m_msgDescStack[this->m_msgDescCursor].profile->key;
		//}
		this->m_msgDescCursor++;
		return true;
	}
	else
	{	
		PERFCOUNTER("PollMessageDevice::RecvNetMessage::poll");

		this->m_msgDescCursor = this->m_msgDescCount = 0;
		this->m_rootProfile->recv_ptr = this->m_rootProfile->recvbuf;
		this->m_rootProfile->recv_left = this->m_recvBufLength;
		this->m_rootProfile->send_ptr = this->m_rootProfile->sendbuf;
		this->m_rootProfile->send_left = this->m_sendBufLength;

		int event_count = 0;
		if ((event_count = epoll_wait(this->m_pollHandle, this->m_eventBuffer, this->m_connections, this->m_waitTimeOut)) > 0)
		{			
			for (int index = 0; index < event_count; index++)
			{
				ProfileList * conn_profile = (ProfileList*)(this->m_eventBuffer[index].data.ptr);
				if (conn_profile != this->m_rootProfile)
				{
					// data received
					if (this->m_eventBuffer[index].events & EPOLLIN)
					{
						int received = recv(conn_profile->sock,  conn_profile->recv_ptr, conn_profile->recv_left, 0);
						if ((received != -1) && (received != 0))
						{
							LOG("%d bytes recv from connection %lu(%d)", received, conn_profile->conn->Id(), conn_profile->sock);

							conn_profile->bytes_recv	+= received;
							conn_profile->recv_ptr		+= received;
							conn_profile->recv_left		-= received;
							while (conn_profile->bytes_recv >= int(sizeof(Internal::PackageHeader)))
							{
								Internal::PackageHeader * header = (Internal::PackageHeader*)conn_profile->proc_ptr;

								if (conn_profile->bytes_recv < header->length)
								{
									if (conn_profile->recv_left < header->length - conn_profile->bytes_recv)
									{
										conn_profile->recv_left = header->length - conn_profile->bytes_recv;
									}
									break;
								}
								else
								{
									//*************************************************************************
									// construct message.
									MessageDescriptor * msgdesc = &(this->m_msgDescStack[this->m_msgDescCount++]);
																		
									msgdesc->profile = conn_profile;
									msgdesc->package.message.command = header->command;
									msgdesc->package.message.catalog = header->catalog;
									msgdesc->package.buffer.data = (char *)(header) + sizeof(Internal::PackageHeader);
									msgdesc->package.buffer.size = header->length;
									msgdesc->package.timestamp	 = time;

									LOG("A message received from Node(%lu[%u]::%d), cmd:%d, seg:%d, size:%lu, time:%lu",
										msgdesc->profile->conn->Id(), header->routine, msgdesc->profile->sock,
										msgdesc->package.message.command, msgdesc->package.message.segment, 
										msgdesc->package.buffer.size, msgdesc->package.timestamp);
									//*************************************************************************
									conn_profile->bytes_recv -= sizeof(Internal::PackageHeader) + header->length;
									if (conn_profile->bytes_recv == 0)
									{
										//------------------------------
										//save the flux_in information.
										//conn_profile->flux_in_bytes += conn_profile->proc_ptr - conndesc->recvbuf;
										//this->m_FluxInBytes += conn_profile->proc_ptr - conn_profile->recvbuf;
										//------------------------------

										// reset stream
										conn_profile->recv_ptr = conn_profile->recvbuf;
										conn_profile->proc_ptr = conn_profile->recv_ptr;
										conn_profile->recv_left = this->m_recvBufLength;
										break;
									}
									conn_profile->proc_ptr += sizeof(Internal::PackageHeader) + header->length;
								}
							}						
						}
						else
						{
							// connection breaked or error occured
							//++++++++++++++++++++++++++++++++++++++++++++
							this->m_eventBuffer[index].events |= EPOLLERR;
							//++++++++++++++++++++++++++++++++++++++++++++
						}
					}
				
					if (this->m_eventBuffer[index].events & EPOLLOUT)
					{
						if (conn_profile->send_ptr > conn_profile->sendbuf)
						{
							char * mark = conn_profile->sendbuf + conn_profile->bytes_sent;
							int sent = send(conn_profile->sock, mark, int(conn_profile->send_ptr - mark), 0);
							if (sent != -1)
							{
								conn_profile->bytes_sent += sent;
								if (conn_profile->bytes_sent == this->m_sendBufLength - conn_profile->send_left)
								{
									//------------------------------
									//save the flux_out information.
									//conndesc->flux_out_bytes += conndesc->bytes_sent;
									//this->m_FluxOutBytes += conndesc->bytes_sent;
									//------------------------------

									LOG("%d bytes send to connection %lu(%d)", sent, conn_profile->conn->Id(), conn_profile->sock);
											
									// reset stream
									conn_profile->bytes_sent = 0;
									conn_profile->send_ptr = conn_profile->sendbuf;
									conn_profile->send_left = this->m_sendBufLength;

									//@ !!
									//=================================================================================
									epoll_event event;
									event.data.ptr = conn_profile;
									event.events = EPOLLIN | EPOLLERR;									
									epoll_ctl(this->m_pollHandle, EPOLL_CTL_MOD, conn_profile->sock, &event);
									//=================================================================================
								}
							}
						}
					}
				
					if (this->m_eventBuffer[index].events & EPOLLERR)
					{
						this->m_hostKernel->OutputConsole("Connection %d broken.", conn_profile->conn->Id());

						close(conn_profile->sock);
						conn_profile->sock = -1;

						conn_profile->next = NULL;
						this->m_profileTail->next = conn_profile;
						this->m_profileTail = this->m_profileTail->next;

						// connection breaked or error occured
						//++++++++++++++++++++++++++++++++++++++++++++						
						SAF::KERNEL::INTERNALCOMMAND::CONNECTIONHANDLE_tag * handletag 
							= (SAF::KERNEL::INTERNALCOMMAND::CONNECTIONHANDLE_tag *)this->m_rootProfile->send_ptr;
						handletag->connectionHandle = (SAF::IConnection *)conn_profile->key;

						MessageDescriptor * msgdesc = &(this->m_msgDescStack[this->m_msgDescCount++]);

						msgdesc->profile = this->m_rootProfile;
						msgdesc->package.message.catalog = 0;
						msgdesc->package.message.command = SAF::KERNEL::INTERNALCOMMAND::CONNECTIONBROKEN;							
						msgdesc->package.buffer.data = this->m_rootProfile->send_ptr;
						msgdesc->package.buffer.size = sizeof(SAF::KERNEL::INTERNALCOMMAND::CONNECTIONHANDLE_tag);
						msgdesc->package.timestamp	 = time;

						this->m_rootProfile->send_ptr += sizeof(SAF::KERNEL::INTERNALCOMMAND::CONNECTIONHANDLE_tag);
						this->m_rootProfile->send_left -= sizeof(SAF::KERNEL::INTERNALCOMMAND::CONNECTIONHANDLE_tag);
						//++++++++++++++++++++++++++++++++++++++++++++
					}			
				}
				else	// conn_profile == this->m_rootProfile ...
				{
					// connection comes					
					sockaddr addr; socklen_t addrlen = sizeof(addr);
					int newsocket = accept(this->m_rootProfile->sock, &addr, &addrlen);
					if (newsocket != -1)
					{						
						ProfileList * alloced_profile = this->m_profileHead;
						if (alloced_profile != NULL)
						{
							this->m_profileHead = alloced_profile->next;							
							alloced_profile->next = NULL;

							alloced_profile->conn = NULL; //((char*)alloced_profile - (char*)this->m_profileBase)/sizeof(ProfileList);
							alloced_profile->sock = newsocket;
							alloced_profile->recv_ptr = alloced_profile->recvbuf;
							alloced_profile->send_ptr = alloced_profile->sendbuf;
							alloced_profile->proc_ptr = alloced_profile->recvbuf;
							alloced_profile->recv_left = this->m_recvBufLength;
							alloced_profile->send_left = this->m_sendBufLength;

							epoll_event event;
							event.data.ptr = alloced_profile;
							event.events = EPOLLIN | EPOLLERR;							
							if (epoll_ctl(this->m_pollHandle, EPOLL_CTL_ADD, newsocket, &event) == 0)
							{
								SAF::KERNEL::INTERNALCOMMAND::CONNECTIONJOIN_tag * jointag 
									= (SAF::KERNEL::INTERNALCOMMAND::CONNECTIONJOIN_tag *)this->m_rootProfile->recv_ptr;
								jointag->connectionId = alloced_profile->conn->Id();
								strncpy(jointag->connectionAddr, inet_ntoa(((sockaddr_in*)&addr)->sin_addr), sizeof(jointag->connectionAddr));
								jointag->connectionPort = ntohs(((sockaddr_in*)&addr)->sin_port);									

								MessageDescriptor * msgdesc = &(this->m_msgDescStack[this->m_msgDescCount++]);
																		
								msgdesc->profile = this->m_rootProfile;
								msgdesc->package.message.catalog = 0;
								msgdesc->package.message.command = SAF::KERNEL::INTERNALCOMMAND::CONNECTIONJOIN;							
								msgdesc->package.buffer.data = this->m_rootProfile->recv_ptr;
								msgdesc->package.buffer.size = sizeof(SAF::KERNEL::INTERNALCOMMAND::CONNECTIONJOIN_tag);
								msgdesc->package.timestamp	 = time;							

								this->m_rootProfile->recv_ptr += sizeof(SAF::KERNEL::INTERNALCOMMAND::CONNECTIONJOIN_tag);
								this->m_rootProfile->recv_left -= sizeof(SAF::KERNEL::INTERNALCOMMAND::CONNECTIONJOIN_tag);


								LOG("Connection %lu created.", alloced_profile->nodeid);
							}
							else
							{
								// failed, rollback
								close(newsocket);
								this->m_profileHead = alloced_profile;									
							}
						}
						else	// connection full
						{
							close(newsocket);							
						}
					}	
					// accept failed ...
				}	// is listen socket ...
			}	// for (int index = 0; index < event_count; ...
		}	// if ((event_count = epoll_wait(this->m_pollHandle, ...)  successful ...		
	}	// if (this->m_msgDescCursor == this->m_msgDescCount) ...
	return false;
}

//bool PollMessageRouter::BindConnection(unsigned long nodeid, void * key)
//{
//	PERFCOUNTER("PollMessageDevice::BindConnection");
//
//	ProfileList * profile = &(this->m_profileBase[nodeid]);
//	assert(profile->nodeid == nodeid);
//	if (profile->nodeid == nodeid)
//	{
//		profile->key = key;
//		return true;
//	}
//	assert(!"connection not found.");
//	return false;
//}


bool EPollMessageRouter::DropConnection(const SAF::IConnection * conn)
{
	PERFCOUNTER("PollMessageDevice::DropConnection");

	ProfileList * profile = &(this->m_profileBase[conn->Id()]);
	//assert(profile->nodeid == conn->Id());
	//if (profile->nodeid == conn->Id())
	{
		close(profile->sock);
		return true;
	}
	assert(!"connection not found.");
	return false;
}


